package main

import (
	"bufio"
	"log"
	"os"
	"strconv"
	"strings"

	"fmt"

	"github.com/optiopay/kafka"
	"github.com/optiopay/kafka/proto"
)

func createConsume(broker kafka.Client, topic string, partition int, offset int64) kafka.Consumer {
	conf := kafka.NewConsumerConf(topic, int32(partition))
	conf.StartOffset = offset
	consumer, err := broker.Consumer(conf)
	if err != nil {
		panic(err)
	}
	return consumer
}
func createMx(broker *kafka.Broker) *kafka.Mx {
	partitionNum, err := broker.PartitionCount(topic)
	if err != nil {
		panic(err.Error())
	}
	if partitionNum == 0 {
		return nil
	}
	var consumers []kafka.Consumer
	for i := int32(0); i != partitionNum; i++ {
		consumers = append(consumers, createConsume(broker, topic, int(i), int64(offset)))
	}
	return kafka.Merge(consumers...)
}

// printConsumed read messages from kafka and print them out
func printConsumed(broker *kafka.Broker) {
	mx := createMx(broker)
	defer mx.Close()
	for {
		msg, err := mx.Consume()
		if err != nil {
			if err != kafka.ErrNoData {
				log.Printf("cannot consume %q topic message: %s", topic, err)
			}
			break
		}
		for i := 0; i != mx.Workers(); i++ {
			offsetLatest, err := broker.OffsetLatest(topic, int32(i))
			if err != nil {
				panic(err.Error())
			}
			log.Printf("partition:[%d] - OffsetLatest:[%d]", i, offsetLatest)
		}
		log.Printf("[%d]-[%d]-[%s]", msg.Partition, msg.Offset, msg.Value)
	}

	log.Print("{consumer quit}")
}

// produceStdin read stdin and send every non empty line as message
func produceStdin(broker kafka.Client) {
	producer := broker.Producer(kafka.NewProducerConf())
	input := bufio.NewReader(os.Stdin)
	scanner := bufio.NewScanner(input)
	for scanner.Scan() {
		line := strings.TrimSpace(scanner.Text())
		if line == "" {
			continue
		}
		lineSegs := strings.Split(line, ",")
		partition, err := strconv.Atoi(lineSegs[0])
		if err != nil {
			fmt.Println("Invalid partion, Please re-type-in")
			continue
		}
		msg := &proto.Message{Value: []byte(lineSegs[1])}
		if _, err := producer.Produce(topic, int32(partition), msg); err != nil {
			log.Fatalf("cannot produce message to %s:%d: %s", topic, partition, err)
		}
	}
}

func main() {
	conf := kafka.NewBrokerConf("test-client")
	conf.AllowTopicCreation = true

	// connect to kafka cluster
	broker, err := kafka.Dial(kafkaAddrs, conf)
	if err != nil {
		log.Fatalf("cannot connect to kafka cluster: %s", err)
	}
	broker.Close()

	go printConsumed(broker)
	produceStdin(broker)
}
